# -*- coding: utf-8 -*-
'''
@datetime : 2022/6/9 18:35
@author   : zhangxp
@file     : tasks.py
'''
from requests import request
from .celery import app

import sys
import os
import re
import time
import socket
from celery import group
from rqalpha import run_file, run_code

from copy import deepcopy
from pytz import timezone
from sqlalchemy import DATETIME, DATE, VARCHAR, create_engine
import numpy as np
import pandas as pd

from concurrent.futures import ThreadPoolExecutor,wait,ALL_COMPLETED
from multiprocessing.process import current_process

##########################################通用任务##################################################
def method_caller(mod, method_name, *args):
    obj = __import__(mod, fromlist=True)
    if hasattr(obj, method_name):
        func = getattr(obj, method_name)
        return func(*args)
    else:
        raise NameError("There is no attr '%s' " % method_name, "in %s.py" % mod)


@app.task(autoretry_for=(Exception,))
def task_gpu(mod, method_name, method_param=None):
    try:
      old_cfg = current_process()._config['daemon']
      current_process()._config['daemon'] = False
      result = method_caller(mod, method_name, *method_param)
      if isinstance(result, pd.DataFrame):
        return result.to_dict('list')
      return result
    finally:
      current_process()._config['daemon'] = old_cfg
 
    
def group_params_gpu_task(func, params_list):
    """
    分组处理回测任务

    :param func: 策略源码的字符串，可以规避分布式节点找到文件问题
    :param params_list: 分组参数，即，将被不通节点执行的回测策略附加参数
    :example:

    .. code-block:: python


    """
    if isinstance(func, tuple):
      mod = func[1]
      method_name = func[0]
    else:
      mod = func.__module__
      method_name = func.__name__
    task_list = []
    for params in params_list:
      # 分布式异步，文件模式
      task_list.append(task_gpu.s(mod, method_name, params))
    # 按组执行
    job = group(task_list)
    result = job.apply_async()
    result.join()
    return {'group_id':result.id, 'status':result.successful(), 'result':result.get()}